package com.mq.mqClient;

import com.mq.common.*;
import lombok.Data;
import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 一个客户端可以有多个连接
 * 一个连接对应多个channel,因此要用rid来区分请求和对应的相应
 */
public class Connection {
    private Socket socket = null;

    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;
    //创建线程池,用来处理客户端这边执行用户回调的线程池
    private ExecutorService callbackPool = null;

    //  创建一个hash.来管理多个channel
    ConcurrentHashMap<String,Channel>  channelMap = new ConcurrentHashMap<>();

    public Connection(String host,int port) throws IOException {
        socket = new Socket(host,port);
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);

        callbackPool = Executors.newFixedThreadPool(4);
//      创建一个扫描线程,不断的从socket中读取响应,交给对应的channel进行处理
        Thread t = new Thread(()->{
           try{
               while (!socket.isClosed()){
                   Response response = readResponse();
                   //处理响应
                   dispatchResponse(response);
               }
           } catch (SocketException e){
              //连接正常断开
               System.out.println("[Connection] 连接正常断开");
           }catch (IOException | ClassNotFoundException | MqException e) {
               System.out.println("[Connection] 连接异常断开");
               e.printStackTrace();
           }
        });
        t.start();
    }

    public void close(){
        try{
        //关闭Connection ,释放资源
            callbackPool.shutdownNow();
            channelMap.clear();
            outputStream.close();
            inputStream.close();
            socket.close();;
        }catch (IOException e){
            e.printStackTrace();
        }
    }


    //处理响应,要判断当前响应是是针对请求返回的响应还是 服务器推送的消息
    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        byte[] payload = response.getPayload();
        //服务器推送的消息,找到对应的channel,执行回调
        if(response.getType()==0xc){
            //1.接收响应结果
            SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.fromBytes(payload);
            //2.找到对应的channel对象
            String channelId = subScribeReturns.getChannelId();
            Channel channel = channelMap.get(channelId);
            if(channel==null){
                throw new MqException("[Connection] 消息对应的channel在客户端中不存在 channelId:"+channelId);
            }
            //执行该channel内部的回调
            callbackPool.submit(()->{
                try {
                    channel.getConsumer().handlerDeliver(subScribeReturns.getConsumerTag(),
                            subScribeReturns.getBasicProperties(),subScribeReturns.getBody());
                } catch (MqException | IOException e) {
                    e.printStackTrace();
                }
            });
        }else{
            //针对请求,服务器返回的响应
            BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(payload);
            //把结果放到对应的channel的哈希表中
            Channel channel = channelMap.get(basicReturns.getChannelId());
            if(channel==null){
                throw new MqException("[Connection] 该消息对应的channel在客户端中不存在! channelId:"+channel.getChannelId());
            }
            channel.putReturnsMap(basicReturns);
        }

    }

//  客户端发送请求给服务器
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        //刷新缓存区
        dataOutputStream.flush();
        System.out.println("[Connection] 发送请求成功 type:"+request.getType()
                +" ,length:"+request.getLength());
    }

    //读取响应
    public Response readResponse() throws IOException{
        Response response = new Response();
        response.setType(dataInputStream.readInt());
        response.setLength(dataInputStream.readInt());
        byte[] payload = new byte[response.getLength()];
        int n = dataInputStream.read(payload);
        if(n!=response.getLength()){
            throw new IOException("[Connection] 读取响应格式有误! ");
        }
        response.setPayload(payload);
        System.out.println("[Connection] 收到响应 type:"+response.getType()
        +" ,length:"+response.getLength());
        return response;
    }

    //在connection中创建一个channel
    public Channel createChannel() throws IOException {
        String channelId = "C-"+ UUID.randomUUID();
        Channel channel = new Channel(channelId, this);
        channelMap.put(channelId,channel);
        //把这个 channel 对象放到 Connection 管理 channel 的 哈希表 中.????????
        boolean ok = channel.createChannel();
        if(!ok){
            //创建失败时,直接返回
            channelMap.remove(channelId,channel);
            return null;
        }
        return channel;
    }

}
